package com.auok.Entity;

import Util.JsonUtil;
import Util.SQLUtil;
import com.auok.Command.CommandHandler;

import com.auok.Message.SendToUserMessage;
import io.vertx.core.json.JsonArray;
import io.vertx.reactivex.core.http.ServerWebSocket;
import io.vertx.reactivex.ext.sql.SQLClient;
import lombok.Getter;

import java.util.stream.Collectors;

@Getter

public class UserSession {
    private User user;
    private ServerWebSocket webSocket;


    public UserSession(User user, ServerWebSocket webSocket, SQLClient sqlClient) {
        this.user = user;
        this.webSocket = webSocket;

        this.webSocket.textMessageHandler(new CommandHandler(this, sqlClient));

        ProcessUnreceivedMessage(sqlClient, user);
    }

    /**
     * 处理未接收的消息
     */
    private void ProcessUnreceivedMessage(SQLClient sqlClient, User user) {

        String querySql = "select * from user_message where data->>'received'='false' and data->>'toUserId'=?";

        JsonArray param = new JsonArray().add(user.getId());

        SQLUtil.doInConnection(sqlClient,
                conn -> conn.rxQueryWithParams(querySql, param)
                        .map(resultSet -> resultSet.getRows().stream()
                                .map(json ->
                                        JsonUtil.FromJson(
                                                json.getString("data")
                                                , UserMessage.class))
                                .collect(Collectors.toList())
                        )
        ).map(list -> {

                    list.forEach(item ->
                            new SendToUserMessage(item.getTime()
                                    , item.getFromUserId()
                                    , item.getContent())
                                    .Send(this));


                    String updateSql = "update user_message set data=jsonb_set(data,'{received}','true') where data->>'received'='false' and data->>'toUserId'=?";

                    return SQLUtil.doInConnection(sqlClient,
                            conn2 -> conn2.rxUpdateWithParams(updateSql, param)
                    ).subscribe();
                }
        ).subscribe();
    }
}
